package org.example;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.errors.TopicExistsException;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 *@description: 创建主题
 *@author: xukang
 *@create: 2025-05-30
 */
public class CreateTopicExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        try (AdminClient adminClient = AdminClient.create(props)) {
            // 构建主题配置
//            NewTopic newTopic = TopicBuilder.name("demo-topic")
//                    .partitions(3)
//                    .replicas(3)  // 生产环境建议至少3个副本
//                    .config(TopicConfig.RETENTION_MS_CONFIG, "86400000")  // 保留1天
//                    .config(TopicConfig.SEGMENT_BYTES_CONFIG, "1073741824")  // 1GB/段
//                    .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "lz4")  // 压缩类型
//                    .build();
            NewTopic newTopic = new NewTopic(
                    "my-topic",      // 主题名称
                    3,               // 分区数
                    (short) 1        // 副本数
            );
            // 创建主题
            CreateTopicsResult result = adminClient.createTopics(Collections.singletonList(newTopic));
            // 等待创建完成
            result.all().get();
            System.out.println("主题创建成功: my-topic");

        } catch (ExecutionException e) {
            if (e.getCause() instanceof TopicExistsException) {
                System.out.println("主题已存在: my-topic");
            } else {
                System.err.println("创建主题失败: " + e.getMessage());
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("操作被中断: " + e.getMessage());
        } catch (Exception e) {
            System.err.println("未知异常: " + e.getMessage());
        }
    }
}
